package com.gvtv.main.hadoop.yarn;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ApplicationReport;
import org.apache.hadoop.yarn.api.records.YarnApplicationState;
import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.ApplicationNotFoundException;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.util.ConverterUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class YarnHelper {

private static final Logger logger = LoggerFactory.getLogger(YarnHelper.class);
	
	private Configuration conf;
	private YarnClient yarnClient = YarnClient.createYarnClient();
	
	public YarnHelper() {
		this.conf = new YarnConfiguration();
		init();
	}
	public YarnHelper(Configuration conf) {
		this.conf = new YarnConfiguration(conf);
		init();
	}
	
	public void init(){
		yarnClient.init(conf);
	    yarnClient.start();
	}
	public List<ApplicationReport> getApps(){
		try {
			return yarnClient.getApplications();
		} catch (YarnException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}
	public List<ApplicationReport> getAppByName(String appName){
		if(getApps()==null){
			return null;
		}
		List<ApplicationReport> list = new ArrayList<ApplicationReport>();
		for(ApplicationReport report : getApps()){
			if(appName.equals(report.getName())){
				list.add(report);
			}
		}
		return list;
	}
	public void killApp(String applicationId) throws YarnException{
		ApplicationId appId = ConverterUtils.toApplicationId(applicationId);
	    ApplicationReport  appReport = null;
	    try {
	      appReport = yarnClient.getApplicationReport(appId);
	    } catch (ApplicationNotFoundException e) {
	    	String msg = "Application with id '"+ applicationId + "' doesn't exist in RM.";
	    	logger.warn(msg);
	      throw new YarnException(msg);
	    } catch (IOException e) {
			throw new YarnException(e);
		} 

	    if (appReport.getYarnApplicationState() == YarnApplicationState.FINISHED
	        || appReport.getYarnApplicationState() == YarnApplicationState.KILLED
	        || appReport.getYarnApplicationState() == YarnApplicationState.FAILED) {
	    	String msg = "Application " + applicationId + " has already finished ";
	    	throw new YarnException(msg);
	    } else {
	    	try {
				yarnClient.killApplication(appId);
			} catch (IOException e) {
				throw new YarnException(e);
			}
	    }
	}
	public void close(){
		yarnClient.stop();
		try {
			yarnClient.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
	public void setConfValue(String name, String value){
		this.conf.set(name, value);
	}
	
	public String getConfValue(String name){
		return getConf().get(name);
	}
	
	public Configuration getConf() {
		return yarnClient.getConfig();
	}
	public void setConf(Configuration conf) {
		this.conf = conf;
	}
	
	public static void main(String[] args) {
		YarnHelper yh = new YarnHelper();
		System.out.println(yh.getApps().size());
	}

}
